# @Time : 2024/6/13 14:50
# @Author : ZHUYI
# @File : consumer1_mysql
from kafka import KafkaConsumer
import json
import csv


def consumer2(topic):
    print("consumer2开始消费。。。。")
    # 订阅主题，使用本地Kafka服务器，从最早的消息开始消费
    consumer = KafkaConsumer(topic, bootstrap_servers=['localhost:9092'],
                             group_id=None, auto_offset_reset='earliest')

    for msg in consumer:
        msg1 = str(msg.value, encoding="utf-8")
        # 两次json.loads处理嵌套的JSON字符串
        json_dict = json.loads(json.loads(msg1))
        rows = []
        # 插入数据
        for d in json_dict:
            for k, v in d.items():
                # 当键为"data"时，进一步处理
                if k == "data":
                    for i in v:
                        # 将字典转换为JSON字符串再转为字典，纠正格式
                        value = json.loads(json.dumps(i))
                        # 提取相关信息
                        base = value['base_info']
                        user = value['user_info']
                        # 掘金小册的具体信息
                        id = base['booklet_id']
                        title = base['title']
                        summary = base['summary']
                        author = user['user_name']
                        job_title = user['job_title']
                        cover_img = base['cover_img']
                        price = value['max_discount']['price']
                        buy_count = base['buy_count']
                        is_finished = base['is_finished']

                        book_tuple = (id, title, summary, author, job_title, cover_img, price, buy_count, is_finished)
                        rows.append(book_tuple)
        # 插入csv文件
        with open('csv/book.csv', 'w', encoding='utf8', newline='') as f:
            # 标题
            headers = ['掘金小册id', '小册标题', '小册介绍', '小册作者', '作者头衔', '小册图片链接',
                       '小册价格', '销售数量', '是否完结']
            writer = csv.writer(f)
            writer.writerow(headers)
            print(rows)
            # 插入数据
            writer.writerows(rows)
        print('成功插入csv')
        consumer.close()
    print("consumer2结束消费。。。。")
